#include "buffer_pool_manager.h"

#include <assert.h>    // for assert
#include <string.h>    // for memset
#include <sys/stat.h>  // for stat, fstat
#include <unistd.h>    // for lseek, read, write, close, unlink
#include <algorithm>   // for std::min

#include "defs.h"      // Assuming PAGE_SIZE, INVALID_PAGE_ID are defined here
#include "errors.h"    // For RedBaseError and its derivatives
#include "page.h"      // For Page class, PageId, PageIdHash

// DiskManager::DiskManager() { memset(fd2pageno_, 0, MAX_FD * (sizeof(std::atomic<page_id_t>) / sizeof(char))); }
// Note: The above line seems to be from DiskManager's constructor and should not be in BufferPoolManager.cpp.
// It's commented out as it causes a redefinition error if DiskManager's constructor is also in its own .cpp file.

/**
 * @brief ´Ófree_list»òreplacerÖÐµÃµ½¿ÉÌÔÌ­Ö¡Ò³µÄ *frame_id
 * @param frame_id Ö¡Ò³idÖ¸Õë,·µ»Ø³É¹¦ÕÒµ½µÄ¿ÉÌæ»»Ö¡id
 * @return true: ¿ÉÌæ»»Ö¡²éÕÒ³É¹¦ , false: ¿ÉÌæ»»Ö¡²éÕÒÊ§°Ü
 */
bool BufferPoolManager::FindVictimPage(frame_id_t *frame_id) {
    // 1 Ê¹ÓÃBufferPoolManager::free_list_ÅÐ¶Ï»º³å³ØÊÇ·ñÒÑÂúÐèÒªÌÔÌ­Ò³Ãæ
    // 1.1 Î´Âú»ñµÃframe
    if (!free_list_.empty()) {
        *frame_id = free_list_.front();
        free_list_.pop_front();
        return true;
    }
    // 1.2 ÒÑÂúÊ¹ÓÃlru_replacerÖÐµÄ·½·¨Ñ¡ÔñÌÔÌ­Ò³Ãæ
    // Èç¹û free_list_ Îª¿Õ£¬Ôò´Ó replacer ÖÐ»ñÈ¡ÎþÉüÖ¡
    return replacer_->Victim(frame_id);
}

/**
 * @brief ¸üÐÂÒ³ÃæÊý¾Ý, ÎªÔàÒ³ÔòÐèÐ´Èë´ÅÅÌ£¬¸üÐÂpageÔªÊý¾Ý(data, is_dirty, page_id)ºÍpage table
 *
 * @param page Ð´»ØÒ³Ö¸Õë
 * @param new_page_id Ð´»ØÒ³ÐÂpage_id
 * @param new_frame_id Ð´»ØÒ³ÐÂÖ¡frame_id
 */
void BufferPoolManager::UpdatePage(Page *page, PageId new_page_id, frame_id_t new_frame_id) {
    // 1 Èç¹ûÊÇÔàÒ³£¬Ð´»Ø´ÅÅÌ£¬²¢ÇÒ°ÑdirtyÖÃÎªfalse
    if (page->is_dirty_) { // Ö±½Ó·ÃÎÊ Page µÄ is_dirty_ ³ÉÔ±
        // È·±£ page_id ÊÇÓÐÐ§µÄ£¬·ñÔòÎÞ·¨Ð´Èë´ÅÅÌ
        if (page->id_.page_no != INVALID_PAGE_ID) { // Ö±½Ó·ÃÎÊ Page µÄ id_ ³ÉÔ±
            disk_manager_->write_page(page->id_.fd, page->id_.page_no, page->data_, PAGE_SIZE); // Ö±½Ó·ÃÎÊ Page µÄ data_ ³ÉÔ±
        } else {
            // Èç¹ûÊÇÔàÒ³µ« page_id ÎÞÐ§£¬Õâ¿ÉÄÜÊÇÒ»¸öÂß¼­´íÎó»òÒ»¸öÐÂ´´½¨µÄÒ³ÃæÔÚ±»Ð´ÈëÇ°¾Í±äÔà
            // Ó¦¸ÃÅ×³ö´íÎó»ò¼ÇÂ¼¾¯¸æ
            LOG_WARN("Attempted to write dirty page with INVALID_PAGE_ID to disk.");
        }
    }

    // 2 ¸üÐÂpage table
    // ´Ó¾ÉµÄ page_id ÒÆ³ý
    if (page->id_.page_no != INVALID_PAGE_ID) { // Ö±½Ó·ÃÎÊ Page µÄ id_ ³ÉÔ±
        page_table_.erase(page->id_);
    }
    // ²åÈëÐÂµÄ page_id ºÍ frame_id Ó³Éä
    page_table_[new_page_id] = new_frame_id;

    // 3 ÖØÖÃpageµÄdata£¬¸üÐÂpage id
    page->ResetMemory(); // Çå¿ÕÒ³ÃæÊý¾Ý
    page->id_ = new_page_id; // Ö±½Ó·ÃÎÊ Page µÄ id_ ³ÉÔ±
    page->is_dirty_ = false; // Çå³ýÔàÎ»
    page->pin_count_ = 0;    // ÖØÖÃ pin_count
}

/**
 * Fetch the requested page from the buffer pool.
 * Èç¹ûÒ³±íÖÐ´æÔÚpage_id£¨ËµÃ÷¸ÃpageÔÚ»º³å³ØÖÐ£©£¬²¢ÇÒpin_count++¡£
 * Èç¹ûÒ³±í²»´æÔÚpage_id£¨ËµÃ÷¸ÃpageÔÚ´ÅÅÌÖÐ£©£¬ÔòÕÒ»º³å³Øvictim page£¬½«ÆäÌæ»»Îª´ÅÅÌÖÐ¶ÁÈ¡µÄpage£¬pin_countÖÃ1¡£
 * @param page_id id of page to be fetched
 * @return the requested page
 */
Page *BufferPoolManager::FetchPage(PageId page_id) {
    // 0. lock latch
    std::scoped_lock lock{latch_};

    // 1. Search the page table for the requested page (P).
    auto it = page_table_.find(page_id);
    if (it != page_table_.end()) {
        // 1.1 If P exists, pin it and return it immediately.
        frame_id_t frame_id = it->second;
        Page *page = &pages_[frame_id];
        page->pin_count_++; // Ö±½Ó·ÃÎÊ Page µÄ pin_count_ ³ÉÔ±
        replacer_->Pin(frame_id); // ¹Ì¶¨¸ÃÖ¡£¬·ÀÖ¹±»ÌÔÌ­
        return page;
    }

    // 1.2 If P does not exist, find a replacement page (R) from either the free list or the replacer.
    //     Note that pages are always found from the free list first.
    frame_id_t victim_frame_id = -1;
    if (!FindVictimPage(&victim_frame_id)) {
        // Èç¹ûÃ»ÓÐ¿ÉÓÃµÄ¿ÕÏÐÖ¡Ò²Ã»ÓÐ¿ÉÌÔÌ­µÄÎþÉüÖ¡£¬Ôò·µ»Ø nullptr
        return nullptr;
    }

    Page *victim_page = &pages_[victim_frame_id];

    // 2. If R is dirty, write it back to the disk.
    if (victim_page->is_dirty_) { // Ö±½Ó·ÃÎÊ Page µÄ is_dirty_ ³ÉÔ±
        // È·±£ÎþÉüÒ³µÄ page_id ÊÇÓÐÐ§µÄ£¬·ñÔòÎÞ·¨Ð´Èë´ÅÅÌ
        if (victim_page->id_.page_no != INVALID_PAGE_ID) { // Ö±½Ó·ÃÎÊ Page µÄ id_ ³ÉÔ±
            disk_manager_->write_page(victim_page->id_.fd, victim_page->id_.page_no, victim_page->data_, PAGE_SIZE); // Ö±½Ó·ÃÎÊ Page µÄ data_ ³ÉÔ±
        } else {
            // ÕâÍ¨³£²»Ó¦¸Ã·¢Éú£¬³ý·ÇÊÇÐÂ·ÖÅäµ«´ÓÎ´Ð´Èë´ÅÅÌµÄÔàÒ³
            LOG_WARN("Attempted to write dirty victim page with INVALID_PAGE_ID to disk.");
        }
    }

    // 3. Delete R from the page table and insert P.
    // ´ÓÒ³±íÖÐÒÆ³ýÎþÉüÒ³µÄ¾ÉÓ³Éä
    if (victim_page->id_.page_no != INVALID_PAGE_ID) { // Ö±½Ó·ÃÎÊ Page µÄ id_ ³ÉÔ±
        page_table_.erase(victim_page->id_);
    }
    // ²åÈëÐÂÒ³ÃæµÄÓ³Éä
    page_table_[page_id] = victim_frame_id;

    // 4. Update P's metadata, read in the page content from disk, and then return a pointer to P.
    victim_page->id_ = page_id; // Ö±½Ó·ÃÎÊ Page µÄ id_ ³ÉÔ±
    victim_page->pin_count_ = 1; // Ö±½Ó·ÃÎÊ Page µÄ pin_count_ ³ÉÔ±
    victim_page->is_dirty_ = false; // ¸Õ´Ó´ÅÅÌ¶ÁÈë£¬²»ÊÇÔàÒ³
    victim_page->ResetMemory();      // Çå¿Õ¾ÉÊý¾Ý£¬×¼±¸¶ÁÈëÐÂÊý¾Ý

    // ´Ó´ÅÅÌ¶ÁÈ¡Ò³ÃæÄÚÈÝ
    disk_manager_->read_page(page_id.fd, page_id.page_no, victim_page->data_, PAGE_SIZE); // Ö±½Ó·ÃÎÊ Page µÄ data_ ³ÉÔ±

    replacer_->Pin(victim_frame_id); // ¹Ì¶¨ÐÂ¶ÁÈëµÄÒ³Ãæ

    return victim_page;
}

/**
 * Unpin the target page from the buffer pool. È¡Ïû¹Ì¶¨pin_count>0µÄÔÚ»º³å³ØÖÐµÄpage
 * @param page_id id of page to be unpinned
 * @param is_dirty true if the page should be marked as dirty, false otherwise
 * @return false if the page pin count is <= 0 before this call, true otherwise
 */
bool BufferPoolManager::UnpinPage(PageId page_id, bool is_dirty) {
    // 0. lock latch
    std::scoped_lock lock{latch_};

    // 1. try to search page_id page P in page_table_
    auto it = page_table_.find(page_id);
    if (it == page_table_.end()) {
        // 1.1 PÔÚÒ³±íÖÐ²»´æÔÚ return false
        return false;
    }

    frame_id_t frame_id = it->second;
    Page *page = &pages_[frame_id];

    // 1.2 PÔÚÒ³±íÖÐ´æÔÚ ÈçºÎ½â³ýÒ»´Î¹Ì¶¨(pin_count)
    if (page->pin_count_ <= 0) { // Ö±½Ó·ÃÎÊ Page µÄ pin_count_ ³ÉÔ±
        // Èç¹û pin_count ÒÑ¾­ <= 0£¬ËµÃ÷Ò³ÃæÃ»ÓÐ±»¹Ì¶¨£¬ÎÞ·¨ÔÙ´Î unpin
        return false;
    }

    page->pin_count_--; // Ö±½Ó·ÃÎÊ Page µÄ pin_count_ ³ÉÔ±

    // 2. Ò³ÃæÊÇ·ñÐèÒªÖÃÔà
    if (is_dirty) {
        page->is_dirty_ = true; // Ö±½Ó·ÃÎÊ Page µÄ is_dirty_ ³ÉÔ±
    }

    // Ö»ÓÐµ± pin_count ¼õÉÙµ½ 0 Ê±£¬²ÅÄÜµ÷ÓÃ Replacer::Unpin() ½«Ò³ÃæÌí¼Óµ½Ìæ»»Æ÷ÖÐ
    if (page->pin_count_ == 0) { // Ö±½Ó·ÃÎÊ Page µÄ pin_count_ ³ÉÔ±
        replacer_->Unpin(frame_id);
    }

    return true;
}

/**
 * Flushes the target page to disk. ½«pageÐ´Èë´ÅÅÌ£»²»¿¼ÂÇpin_count
 * @param page_id id of page to be flushed, cannot be INVALID_PAGE_ID
 * @return false if the page could not be found in the page table, true otherwise
 */
bool BufferPoolManager::FlushPage(PageId page_id) {
    // 0. lock latch
    std::scoped_lock lock{latch_};

    // 1. Ò³±í²éÕÒ
    auto it = page_table_.find(page_id);
    if (it == page_table_.end()) {
        return false; // Ò³ÃæÎ´ÕÒµ½
    }

    frame_id_t frame_id = it->second;
    Page *page = &pages_[frame_id];

    // 2. ´æÔÚÊ±ÈçºÎÐ´»Ø´ÅÅÌ
    // Make sure you call DiskManager::WritePage!
    disk_manager_->write_page(page->id_.fd, page->id_.page_no, page->data_, PAGE_SIZE); // Ö±½Ó·ÃÎÊ Page µÄ id_ ºÍ data_ ³ÉÔ±

    // 3. Ð´»ØºóÒ³ÃæµÄÔàÎ»
    page->is_dirty_ = false; // Ð´Èë´ÅÅÌºó£¬ÔàÎ»Çå³ý

    return true;
}

/**
 * Creates a new page in the buffer pool. Ïàµ±ÓÚ´Ó´ÅÅÌÖÐÒÆ¶¯Ò»¸öÐÂ½¨µÄ¿Õpageµ½»º³å³ØÄ³¸öÎ»ÖÃ
 * @param[out] page_id id of created page
 * @return nullptr if no new pages could be created, otherwise pointer to new page
 */
Page *BufferPoolManager::NewPage(PageId *page_id) {
    // 0.   lock latch
    std::scoped_lock lock{latch_};

    // 1.   Make sure you call DiskManager::AllocatePage!
    // ·ÖÅäÒ»¸öÐÂµÄÒ³Ãæ ID
    PageId new_page_id;
    new_page_id.fd = page_id->fd; // Ê¹ÓÃ´«Èë page_id µÄ fd
    new_page_id.page_no = disk_manager_->AllocatePage(new_page_id.fd);
    *page_id = new_page_id; // ÉèÖÃÊä³ö²ÎÊý

    // 2.   If all the pages in the buffer pool are pinned, return nullptr.
    frame_id_t victim_frame_id = -1;
    if (!FindVictimPage(&victim_frame_id)) {
        // Èç¹ûÃ»ÓÐ¿ÉÓÃµÄ¿ÕÏÐÖ¡Ò²Ã»ÓÐ¿ÉÌÔÌ­µÄÎþÉüÖ¡£¬Ôò·µ»Ø nullptr
        return nullptr;
    }

    Page *new_page_frame = &pages_[victim_frame_id];

    // 3.   Pick a victim page P from either the free list or the replacer. Always pick from the free list first.
    // (ÒÑÔÚ FindVictimPage ÖÐ´¦Àí)

    // 4.   Update P's metadata, zero out memory and add P to the page table. pin_count set to 1.
    // Èç¹ûÎþÉüÒ³ÊÇÔàÒ³£¬ÏÈÐ´»Ø´ÅÅÌ
    if (new_page_frame->is_dirty_) { // Ö±½Ó·ÃÎÊ Page µÄ is_dirty_ ³ÉÔ±
        if (new_page_frame->id_.page_no != INVALID_PAGE_ID) { // Ö±½Ó·ÃÎÊ Page µÄ id_ ³ÉÔ±
            disk_manager_->write_page(new_page_frame->id_.fd, new_page_frame->id_.page_no, new_page_frame->data_, PAGE_SIZE); // Ö±½Ó·ÃÎÊ Page µÄ data_ ³ÉÔ±
        } else {
            LOG_WARN("Attempted to write dirty new_page_frame with INVALID_PAGE_ID to disk.");
        }
    }

    // ´ÓÒ³±íÖÐÒÆ³ý¾ÉÓ³Éä
    if (new_page_frame->id_.page_no != INVALID_PAGE_ID) { // Ö±½Ó·ÃÎÊ Page µÄ id_ ³ÉÔ±
        page_table_.erase(new_page_frame->id_);
    }

    // ¸üÐÂÐÂÒ³ÃæµÄÔªÊý¾Ý
    new_page_frame->id_ = new_page_id; // Ö±½Ó·ÃÎÊ Page µÄ id_ ³ÉÔ±
    new_page_frame->pin_count_ = 1; // Ö±½Ó·ÃÎÊ Page µÄ pin_count_ ³ÉÔ±
    new_page_frame->is_dirty_ = false; // ÐÂ´´½¨µÄÒ³Ãæ²»ÊÇÔàÒ³
    new_page_frame->ResetMemory();      // Çå¿ÕÄÚ´æ£¬È·±£ÊÇ¿ÕÒ³Ãæ

    // Ìí¼Óµ½Ò³±í
    page_table_[new_page_id] = victim_frame_id;

    // 5.   Set the page ID output parameter. Return a pointer to P.
    replacer_->Pin(victim_frame_id); // ¹Ì¶¨ÐÂÒ³Ãæ

    return new_page_frame;
}

/**
 * @brief Deletes a page from the buffer pool.
 * @param page_id id of page to be deleted
 * @return false if the page exists but could not be deleted, true if the page didn't exist or deletion succeeded
 */
bool BufferPoolManager::DeletePage(PageId page_id) {
    // 0.   lock latch
    std::scoped_lock lock{latch_};

    // 2.   Search the page table for the requested page (P).
    auto it = page_table_.find(page_id);
    // 2.1  If P does not exist, return true.
    if (it == page_table_.end()) {
        // 1.   Make sure you call DiskManager::DeallocatePage!
        // Èç¹ûÒ³Ãæ²»ÔÚ»º³å³ØÖÐ£¬µ«¿ÉÄÜÔÚ´ÅÅÌÉÏ£¬ÐèÒªµ÷ÓÃ DiskManager::DeallocatePage
        // ¸ù¾ÝÄãÌá¹©µÄ DiskManager::DeallocatePage(page_id_t page_id) ½Ó¿Ú£¬
        // Ó¦¸ÃÔÚ´Ë´¦µ÷ÓÃ£¬µ«ÆäÄÚ²¿ÊÇ¿ÕµÄ£¬ËùÒÔÕâÀïÖ»ÊÇÐÎÊ½ÉÏµÄµ÷ÓÃ¡£
        disk_manager_->DeallocatePage(page_id.page_no); // µ÷ÓÃ DiskManager µÄ DeallocatePage
        return true;
    }

    frame_id_t frame_id = it->second;
    Page *page = &pages_[frame_id];

    // 2.2  If P exists, but has a non-zero pin-count, return false. Someone is using the page.
    if (page->pin_count_ > 0) { // Ö±½Ó·ÃÎÊ Page µÄ pin_count_ ³ÉÔ±
        return false; // Ò³ÃæÕýÔÚ±»Ê¹ÓÃ£¬²»ÄÜÉ¾³ý
    }

    // 3.   Otherwise, P can be deleted. Remove P from the page table, reset its metadata and return it to the free list.
    // Èç¹ûÒ³ÃæÊÇÔàÒ³£¬ÏÈÐ´»Ø´ÅÅÌ (ËäÈ»ÒªÉ¾³ý£¬µ«ÎªÁËÊý¾ÝÒ»ÖÂÐÔ£¬Í¨³£»áÏÈÐ´»Ø)
    if (page->is_dirty_) { // Ö±½Ó·ÃÎÊ Page µÄ is_dirty_ ³ÉÔ±
        disk_manager_->write_page(page->id_.fd, page->id_.page_no, page->data_, PAGE_SIZE); // Ö±½Ó·ÃÎÊ Page µÄ id_ ºÍ data_ ³ÉÔ±
    }

    // ´ÓÒ³±íÖÐÒÆ³ý
    page_table_.erase(page_id);

    // ´Ó replacer ÖÐÒÆ³ý£¨Èç¹ûËüÔÚ replacer ÖÐ£©
    replacer_->Pin(frame_id); // Pin »á½«Ò³Ãæ´Ó replacer ÖÐÒÆ³ý

    // ÖØÖÃÒ³ÃæÔªÊý¾Ý
    page->id_.page_no = INVALID_PAGE_ID; // ±ê¼ÇÎªÎÞÐ§Ò³Ãæ
    page->id_.fd = -1; // ±ê¼ÇÎªÎÞÐ§ÎÄ¼þÃèÊö·û
    page->pin_count_ = 0; // Ö±½Ó·ÃÎÊ Page µÄ pin_count_ ³ÉÔ±
    page->is_dirty_ = false; // Ö±½Ó·ÃÎÊ Page µÄ is_dirty_ ³ÉÔ±
    page->ResetMemory(); // Çå¿ÕÄÚ´æ

    // ·µ»Øµ½ free_list
    free_list_.push_back(frame_id);

    // 1.   Make sure you call DiskManager::DeallocatePage!
    // µ÷ÓÃ DiskManager::DeallocatePage ÊÍ·Å´ÅÅÌÉÏµÄÒ³Ãæ
    // ÔÙ´ÎÇ¿µ÷£¬DiskManager::DeallocatePage Ä¿Ç°ÊÇ¿ÕÊµÏÖ£¬ÕâÀïÖ»ÊÇÐÎÊ½ÉÏµÄµ÷ÓÃ¡£
    disk_manager_->DeallocatePage(page_id.page_no); // µ÷ÓÃ DiskManager µÄ DeallocatePage

    return true;
}

/**
 * @brief Flushes all the pages in the buffer pool to disk.
 *
 * @param fd Ö¸¶¨µÄdiskfile open¾ä±ú
 */
void BufferPoolManager::FlushAllPages(int fd) {
    std::scoped_lock lock{latch_};
    for (size_t i = 0; i < pool_size_; i++) {
        Page *page = &pages_[i];
        // ¼ì²éÒ³ÃæÊÇ·ñÊôÓÚÖ¸¶¨µÄÎÄ¼þ£¬²¢ÇÒÊÇÓÐÐ§Ò³Ãæ
        if (page->id_.fd == fd && page->id_.page_no != INVALID_PAGE_ID) { // Ö±½Ó·ÃÎÊ Page µÄ id_ ³ÉÔ±
            // Ö»ÓÐÔàÒ³²ÅÐèÒªÐ´»Ø´ÅÅÌ£¬µ«ÌâÄ¿ÒªÇó¡°ÎÞÂÛ¸ÃÒ³ÊÇ·ñÎªÔàÒ³£¬¶¼½«ÆäË¢ÐÂµ½´ÅÅÌ¡±
            // ËùÒÔÕâÀïÖ±½ÓÐ´»Ø£¬È»ºóÇå³ýÔàÎ»
            disk_manager_->write_page(page->id_.fd, page->id_.page_no, page->data_, PAGE_SIZE); // Ö±½Ó·ÃÎÊ Page µÄ id_ ºÍ data_ ³ÉÔ±
            page->is_dirty_ = false; // Ö±½Ó·ÃÎÊ Page µÄ is_dirty_ ³ÉÔ±
        }
    }
}

